package com.ad2campaign.engine.rest;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.log4j.Logger;

import com.ad2campaign.engine.db.model.Node;
/*
 * Round Robin implementation of load balancer.
 * Relies on java concurrent classes.
 * Yes this class is thread-safe.
 * */
public class RRLoadBalancer implements Loadbalancer
{ 
  	private List<Node> nodes=new CopyOnWriteArrayList<Node>();
 	private AtomicInteger roundRobinCounter = new AtomicInteger(-1);
 	private ActionExecutor executor= null;
 	private Set<Node> blacklistedNodes = new CopyOnWriteArraySet<Node>();
 	private static final Logger LOGGER = Logger.getLogger(RRLoadBalancer.class);
 	private boolean monitorLink=true;
 	private int maxAttempt = 0;
	private NodeMonitor nodeChecker =null  ;
	private NodesLoader nodesLoader=null;
 	private long checkInterval=5000;
 	private Timer nodeMonitor = null; 
 	private Action pingAction;












	@Override
 	public void init()
 	{
 		nodeChecker= new NodeMonitor(blacklistedNodes, nodes, executor,pingAction);
		refresh();
 		if (nodes!=null)
 		{
 		 			maxAttempt =nodes.size()*2;
 		}
 		else
 		{
 			maxAttempt=0;
 		}

  		if (isMonitorLink())
 		{
 			 nodeMonitor = new Timer("NodeMonitor");
 			 nodeMonitor.scheduleAtFixedRate(nodeChecker, 5000, checkInterval);
 			if (LOGGER.isDebugEnabled())
 				{
 					LOGGER.debug("Monitor is on for loadbalancer");
 				}
 		}
  		
		if (LOGGER.isDebugEnabled())
			{
				LOGGER.debug("Maximum attempt for checking nodes : ["+maxAttempt+"]");
			}

 			 
  	}
 	
	


	@Override
	public RestActionResponse process(Action action,List<RequestParameter> requestParameters ,Map<String, ?> urlVariables ,Set<Node> excluded) 
	{
		RestActionResponse response =null;
		if (nodes==null || nodes.size()==0)
		{
			return RestActionResponse.NO_NODE_AVAILABLE;
 		}
		
	 
		int attempt=0;
		int nextNodeIndex=getNextNode();
		Node nextNode= nodes.get(nextNodeIndex);
		boolean isExcluded =(excluded!=null && excluded.contains(nextNode) );
 		
		
 		//First attempt balanced node
		if (LOGGER.isDebugEnabled())
		{
			LOGGER.debug("RRLoadBalancer.process("+action.toString()+",..,..) : Executing request to " +nextNode );
		}
		if (blacklistedNodes.contains(nextNode)==false && isExcluded==false )
		{ 
			response = executor.execute(nextNode, action,requestParameters,urlVariables);
 		}
		
		if (LOGGER.isDebugEnabled())
			{
				LOGGER.debug("RRLoadBalancer.process("+action.toString()+",..,..) : Got response "+response  );
			}
		
  		while ( (response==null || response.isStatus()==false) &&  attempt <= maxAttempt  )
  		{
  	
  			//Attempt to other node
  			nextNode= nodes.get(getNextNode());
   	
  			isExcluded =(excluded!=null && excluded.contains(nextNode) );
  			
  			if (blacklistedNodes.contains(nextNode) || isExcluded )
				{
  	   				//Increment the attempt.
  					attempt++;
 					continue;
				}
			
  			if (LOGGER.isDebugEnabled())
			{
				LOGGER.debug("RRLoadBalancer.process("+action.toString()+",..,..) : Executing request to " +nextNode  );
			}
   			response =  executor.execute(nextNode, action,requestParameters,urlVariables);
   			if (LOGGER.isDebugEnabled())
   			{
   				LOGGER.debug("RRLoadBalancer.process("+action.toString()+",..,..) : Got response "+response  );
   			}
   			//Increment the attempt.
   			attempt++;
   		}
 	
  		if (response==null)
  		{
  			return RestActionResponse.NO_NODE_REACHABLE;
  		}
   		return response; 
  	
	}

	
	
	public int getNextNode()
	{
		if (roundRobinCounter.intValue()==-1)
		{
			roundRobinCounter.set(0);
			
			return roundRobinCounter.intValue();
		}
		else if (roundRobinCounter.intValue() >= nodes.size()-1)
		{
			 roundRobinCounter.set(0);
			 return roundRobinCounter.intValue();
		}
		else
		{
			return roundRobinCounter.incrementAndGet();
		}
	}
	@Override
	public ActionExecutor getExecutor() 
	{
		return executor;
	}
	
	public void setExecutor(ActionExecutor executor) 
	{
		this.executor = executor;
	}
	
 
  	 
	@Override
 	public boolean isMonitorLink()
 	{
		return monitorLink;
	}

 	@Override
	public void setMonitorLink(boolean monitorLink)
	{
		this.monitorLink = monitorLink;
	}

 	public Set<Node> getBlacklistedNodes() 
 	{
		return blacklistedNodes;
	}
 	public void setBlacklistedNodes(Set<Node> blacklistedNodes) 
	{
		this.blacklistedNodes = blacklistedNodes;
	}

 	@Override
 	public NodesLoader getNodesLoader() {
		return nodesLoader;
	}



 	@Override
	public void setNodesLoader(NodesLoader nodesLoader) {
		this.nodesLoader = nodesLoader;
	}


	@Override
	public void refresh()
	{
		if (nodesLoader !=null)
		{
			List<Node> fetchedNodes = nodesLoader.loadNodes();
			if (fetchedNodes!=null && nodes!=null)
			{
				for (Node node:fetchedNodes)
				{
					if (!nodes.contains(node))
					{
						nodes.add(node);
					}
				}
	 			maxAttempt =nodes.size()*2;
	 		}
		}
	}

	public long getCheckInterval()
	{
		return checkInterval;
	}
 	public void setCheckInterval(long checkInterval) 
	{
		this.checkInterval = checkInterval;
	}
 	
 	@Override
 	public void destroy()
 	{
 		if (nodeMonitor!=null)
 		{
 			try 
 			{
				nodeMonitor.cancel();
			} catch (Exception e) {}
 		}
 	}
 	
	public Action getPingAction() 
	{
		return pingAction;
	}
 	public void setPingAction(Action pingAction) 
	{
		this.pingAction = pingAction;
	}

 
 }



